package org.example.batchInsert;

import java.sql.*;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicLong;

public class MassiveCustomerInsert {

    // 进度跟踪器
    private static final AtomicLong counter = new AtomicLong(0);
    private static final long TOTAL_RECORDS = 1_000_000;
    private static final int BATCH_SIZE = 5000;

    public static void main(String[] args) {
        final String url = "jdbc:mysql://localhost:3306/test5?rewriteBatchedStatements=true&useSSL=false&allowPublicKeyRetrieval=true";
        final String username = "root";
        final String password = "123456";

        // 1. 初始化数据库连接
        try (Connection connection = DriverManager.getConnection(url, username, password)) {
            // 禁用自动提交，使用批处理事务
            connection.setAutoCommit(false);

            // 2. 创建线程池并行处理
            int processors = Runtime.getRuntime().availableProcessors();
            System.out.println("使用 " + processors + " 个线程并行处理...");

            Thread[] threads = new Thread[processors];
            for (int i = 0; i < processors; i++) {
                threads[i] = new Thread(() -> insertBatch(connection), "Insert-Thread-" + i);
                threads[i].start();
            }

            // 3. 等待所有线程完成
            for (Thread thread : threads) {
                thread.join();
            }

            System.out.println("\n数据插入完成！共插入 " + TOTAL_RECORDS + " 条记录");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static void insertBatch(Connection connection) {
        // 每个线程处理的数据量
        final long recordsPerThread = TOTAL_RECORDS / Runtime.getRuntime().availableProcessors();
        final Random random = new Random();

        String sql = "INSERT INTO bo_busi_customer (" +
                "cust_id, cust_type, cust_name, nick_name, cust_email, mobile_phone, " +
                "root_user_id, id_card_type, id_card_no, origin_channel, sale_channel, " +
                "is_agent_cust, industry_id, cust_level, consumption_level, cur_manager_id, " +
                "cur_maintain_id, status_cd, status_date, province_id, city_id, branch_id, " +
                "create_staff, create_date, update_staff, update_date, remark, department_id, " +
                "cust_cloud_type, next_follow_date, is_open_sea, is_main_account, " +
                "register_channel_type, register_channel_name, key_word, latest_clue_time" +
                ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";

        try (PreparedStatement ps = connection.prepareStatement(sql)) {
            long startTime = System.currentTimeMillis();
            long baseTimestamp = startTime;

            for (long recordCount = 0; recordCount < recordsPerThread; ) {
                // 4. 当前批次插入
                for (int batchCount = 0; batchCount < BATCH_SIZE && recordCount < recordsPerThread; batchCount++, recordCount++) {
                    String custId = UUID.randomUUID().toString().replace("-", "");

                    // 设置参数
                    ps.setString(1, custId);  // cust_id
                    ps.setString(2, "T" + random.nextInt(100)); // cust_type
                    ps.setString(3, "客户" + counter.incrementAndGet()); // cust_name
                    ps.setString(4, "昵称" + random.nextInt(10000)); // nick_name
                    ps.setString(5, "user" + counter.get() + "@demo.com"); // cust_email
                    ps.setString(6, "1" + String.format("%09d", random.nextInt(1000000000))); // mobile_phone
                    ps.setString(7, "root" + random.nextInt(100000)); // root_user_id
                    ps.setString(8, "ID" + random.nextInt(10)); // id_card_type
                    ps.setString(9, "IDNO" + random.nextInt(10000000)); // id_card_no
                    ps.setString(10, "ORG" + random.nextInt(20)); // origin_channel
                    ps.setString(11, "SALE" + random.nextInt(20)); // sale_channel
                    ps.setString(12, random.nextBoolean() ? "Y" : "N"); // is_agent_cust
                    ps.setString(13, "IND" + random.nextInt(50)); // industry_id
                    ps.setString(14, "L" + random.nextInt(10)); // cust_level
                    ps.setString(15, "V" + random.nextInt(11)); // consumption_level
                    ps.setString(16, "mgr" + random.nextInt(1000)); // cur_manager_id
                    ps.setString(17, "mtn" + random.nextInt(1000)); // cur_maintain_id
                    ps.setString(18, "S" + random.nextInt(5)); // status_cd
                    ps.setLong(19, baseTimestamp - random.nextInt(100000000)); // status_date
                    ps.setString(20, "PROV" + random.nextInt(34)); // province_id
                    ps.setString(21, "CITY" + random.nextInt(333)); // city_id
                    ps.setString(22, "DIST" + random.nextInt(3000)); // branch_id
                    ps.setString(23, "staff" + random.nextInt(100)); // create_staff
                    ps.setLong(24, baseTimestamp - random.nextInt(300000000)); // create_date
                    ps.setString(25, "staff" + random.nextInt(100)); // update_staff
                    ps.setLong(26, baseTimestamp - random.nextInt(100000000)); // update_date
                    ps.setString(27, "备注" + random.nextInt(10000)); // remark
                    ps.setString(28, "DEPT" + random.nextInt(100)); // department_id
                    ps.setString(29, "CLOUD" + random.nextInt(5)); // cust_cloud_type
                    ps.setLong(30, baseTimestamp + random.nextInt(100000000)); // next_follow_date
                    ps.setString(31, random.nextBoolean() ? "Y" : "N"); // is_open_sea
                    ps.setString(32, random.nextBoolean() ? "Y" : "N"); // is_main_account
                    ps.setString(33, "REG" + random.nextInt(10)); // register_channel_type
                    ps.setString(34, "渠道" + random.nextInt(50)); // register_channel_name
                    ps.setString(35, "KW" + random.nextInt(1000)); // key_word
                    ps.setLong(36, baseTimestamp - random.nextInt(200000000)); // latest_clue_time

                    ps.addBatch();
                }

                // 5. 执行批处理
                ps.executeBatch();
                connection.commit();

                // 6. 进度报告
                long inserted = counter.get();
                if (inserted % 100000 == 0) {
                    double percent = (inserted * 100.0) / TOTAL_RECORDS;
                    long elapsed = (System.currentTimeMillis() - startTime) / 1000;
                    System.out.printf("已插入: %,d 条 (%.1f%%) 耗时: %d秒 速度: %,d 条/秒%n",
                            inserted, percent, elapsed,
                            elapsed > 0 ? inserted / elapsed : inserted);
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
            try {
                connection.rollback();
            } catch (SQLException ex) {
                ex.printStackTrace();
            }
        }
    }

    static {
        try {
            Class.forName("com.mysql.cj.jdbc.Driver");
        } catch (ClassNotFoundException e) {
            throw new RuntimeException("找不到MySQL JDBC驱动", e);
        }
    }
}